package ex.ex2;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 舆情监控系统
 */
public class PublicSentimentDRPCTopology {
    private static final Logger logger = LoggerFactory.getLogger(PublicSentimentDRPCTopology.class);

    public static void main(String[] args) throws Exception {
        String function = "infoResearch";
        String topology = "public-sentiment-drpc";

        // 此处定义函数名
        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(function);

        // 关键字处理
        builder.addBolt(new KeywordBoltA()).shuffleGrouping();
        // 数字处理
        builder.addBolt(new NumberBoltB()).shuffleGrouping();
        // 内容长度处理
        builder.addBolt(new LengthBoltC()).shuffleGrouping();

        Config config = new Config();
        if (args == null || args.length == 0) {
            LocalDRPC localDRPC = new LocalDRPC();
            LocalCluster localCluster = new LocalCluster();

            localCluster.submitTopology(topology, config, builder.createLocalTopology(localDRPC));

            String[] urls = new String[]{
                    "12345 投诉", "新冠疫情", "海边游玩"
            };

            for (String url : urls) {
                String result = localDRPC.execute(function, url);
                logger.info("---------------------------------------" + url + " -- " + result);
            }
            localCluster.shutdown();
            localDRPC.shutdown();
        } else {
            topology = args[0];
            StormSubmitter.submitTopology(topology, config, builder.createRemoteTopology());
        }

    }
}
